package com.atbeijing.D04;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import java.sql.Timestamp;
import java.util.HashSet;

/**
 * 每个小时的uv数据
 * 一个用户点击的多次只算一次
 */
public class Example8 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .readTextFile("C:\\Users\\WangJinGen\\IdeaProjects\\flink\\data\\UserBehavior.csv")
                .map(new MapFunction<String, UserBehavior>() {
                    @Override
                    public UserBehavior map(String value) throws Exception {
                        String[] s = value.split(",");
                        return new UserBehavior(s[0],s[1],s[2],s[3],Long.parseLong(s[4])*1000);
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<UserBehavior>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>(){
                                    @Override
                                    public long extractTimestamp(UserBehavior element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                })
                )
                .windowAll(TumblingEventTimeWindows.of(Time.hours(1)))
                .aggregate(new Agg6(),new RestWindow6())
                .print();


        env.execute();
    }

    public static class Agg6 implements AggregateFunction<UserBehavior, HashSet,HashSet>{
        @Override
        public HashSet createAccumulator() {
            return new HashSet();
        }

        @Override
        public HashSet add(UserBehavior value, HashSet accumulator) {
            accumulator.add(value.userId);
            return accumulator;
        }

        @Override
        public HashSet getResult(HashSet accumulator) {
            return accumulator;
        }

        @Override
        public HashSet merge(HashSet a, HashSet b) {
            return null;
        }
    }

    public static class RestWindow6 extends ProcessAllWindowFunction<HashSet,String, TimeWindow>{
        @Override
        public void process(Context context, Iterable<HashSet> elements, Collector<String> out) throws Exception {
            String windowStart = new Timestamp(context.window().getStart()).toString();
            String windowEnd = new Timestamp(context.window().getEnd()).toString();
            out.collect(windowStart+"~"+windowEnd+"\tUV数据"+elements.iterator().next().size());
        }
    }

    public static class UserBehavior {
        public String userId;
        public String itemId;
        public String categoryId;
        public String behaviorType;
        public Long timestamp;

        public UserBehavior() {
        }

        public UserBehavior(String userId, String itemId, String categoryId, String behaviorType, Long timestamp) {
            this.userId = userId;
            this.itemId = itemId;
            this.categoryId = categoryId;
            this.behaviorType = behaviorType;
            this.timestamp = timestamp;
        }

        @Override
        public String toString() {
            return "UserBehavior{" +
                    "userId='" + userId + '\'' +
                    ", itemId='" + itemId + '\'' +
                    ", categoryId='" + categoryId + '\'' +
                    ", behaviorType='" + behaviorType + '\'' +
                    ", timestamp=" + new Timestamp(timestamp) +
                    '}';
        }
    }
}
